package cn.com.bluemoon.daps.model.service.impl;


import cn.com.bluemoon.daps.common.domain.ResultBean;
import cn.com.bluemoon.daps.common.domain.UserInfoHolder;
import cn.com.bluemoon.daps.common.enums.BmStatus;
import cn.com.bluemoon.daps.common.enums.DatabaseType;
import cn.com.bluemoon.daps.common.enums.SystemType;
import cn.com.bluemoon.daps.common.exception.DapException;
import cn.com.bluemoon.daps.model.dto.ModelInfoDto;
import cn.com.bluemoon.daps.model.dto.score.TableScoreDto;
import cn.com.bluemoon.daps.model.entity.*;
import cn.com.bluemoon.daps.model.mapper.DapDataModelMapper;
import cn.com.bluemoon.daps.model.mapper.DapDataModelSubordinateKeyMapper;
import cn.com.bluemoon.daps.model.mapper.DapDataModelTableMapper;
import cn.com.bluemoon.daps.model.mapper.DapDataModelTablePartitionMapper;
import cn.com.bluemoon.daps.model.service.*;
import cn.com.bluemoon.daps.model.vo.DataModelInsertVo;
import cn.com.bluemoon.daps.model.vo.DataModelPageVo;
import cn.com.bluemoon.daps.model.vo.score.SearchTableScoreVo;
import cn.com.bluemoon.daps.system.entity.DapSystemDatasource;
import cn.com.bluemoon.daps.system.entity.DapSystemInfo;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuple4;
import reactor.util.function.Tuples;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

/**
 * <p>
 * 数据模型-模型表 服务实现类
 * </p>
 *
 * @author admin
 * @since 2020-08-03
 */
@Slf4j
@Service
@Transactional
public class DapDataModelServiceImpl extends ServiceImpl<DapDataModelMapper, DapDataModel> implements DapDataModelService {

    @Resource
    DapDataModelMapper dapDataModelMapper;

    @Resource
    DapDataModelTableMapper dapDataModelTableMapper;

    @Resource
    DapDataModelSystemTagMiddleService dapDataModelSystemTagMiddleService;

    @Resource
    DapDataModelSystemTagMiddleService dst;

    @Autowired
    private DapDataModelUserAccreditService modelUserAccreditService;

    @Resource
    private DapSystemModuleFacade systemModuleFacade;

    @Resource
    private DapDataModelGroupService groupService;

    @Resource
    private DapDataModelShareTableService shareTableService;

    @Resource
    private DapDataModelUserAccreditService dapDataModelUserAccreditService;

    @Resource
    private DapDataModelTableService tableService;

    @Resource
    private DapDataModelTableFieldService fieldService;

    @Resource
    private DapDataModelSubordinateKeyService indexService;

    @Resource
    private DapDataModelTablePartitionService partitionService;

    @Resource
    private DapDataModelRecordService dapDataModelRecordService;

    /**
     * 查询模型列表 v1.2
     *
     * @param vo
     * @return
     */
    @Override
    @Transactional(rollbackFor = {DapException.class, RuntimeException.class})
    public IPage<ModelInfoDto> getModelList(DataModelPageVo vo) {

        Page<DapDataModel> page = new Page<>(vo.getPageNum(), vo.getPageSize());
        IPage<ModelInfoDto> dapDataModelIPage = dapDataModelMapper.selectModelList(page, vo);
        // 模型id集合
        List<String> collect = dapDataModelIPage.getRecords().stream().map(ModelInfoDto::getId).collect(Collectors.toList());
        // 获取拥有模型编辑权限的列表
        if (!collect.isEmpty()) {
            List<DapDataModelUserAccredit> userList = dapDataModelUserAccreditService.getBaseMapper()
                    .selectList(new LambdaQueryWrapper<DapDataModelUserAccredit>()
                            .in(DapDataModelUserAccredit::getModelId, collect)
                            .eq(DapDataModelUserAccredit::getBmStatus, BmStatus.ON.getCode()));
            // 当前登陆用户
            String account = UserInfoHolder.getAccount();
            // 根据模型id进行分组
            Map<String, List<DapDataModelUserAccredit>> map = userList.stream().
                    collect(Collectors.groupingBy(DapDataModelUserAccredit::getModelId, Collectors.toList()));
            dapDataModelIPage.getRecords().forEach(dto -> {
                for (Map.Entry<String, List<DapDataModelUserAccredit>> entry : map.entrySet()) {
                    if (entry.getKey().equals(dto.getId())) {
                        List<String> ids = entry.getValue().stream().map(DapDataModelUserAccredit::getUserAccount).collect(Collectors.toList());
                        dto.setBelongTo(ids.contains(account) ? true : false);
                    }
                }
                // 获得模型下表的表数量
                dto.setTableNum(dapDataModelTableMapper.selectTableNum(dto.getId()));
            });
        }
        return dapDataModelIPage;
    }

    /**
     * 新增模型
     *
     * @param vo
     * @return
     */
    @Override
    @Transactional(rollbackFor = {DapException.class, RuntimeException.class})
    public Boolean insertModel(DataModelInsertVo vo) {

        vo.getDapDataModel().setId(null);
        dapDataModelMapper.insert(vo.getDapDataModel());
        // 向模型关联标签表插入数据
        Boolean aBoolean = dst.insertModelTag(vo.getSysTags(), vo.getDapDataModel().getId(), 0);
        // 初始化模型默认分组
        groupService.initModelDefaultGroup(vo.getDapDataModel().getId());
        // 增加模型管理者的变更操作
        DapDataModel dapDataModel = vo.getDapDataModel();
        if (dapDataModel != null) {
            modelUserAccreditService.changUserAccreditByModelSaveOrUpdate(dapDataModel.getId(), dapDataModel.getModelAdmin());
            // 2021年11月9日11:13:49 创建人可以默认增加具有权限操作该模型
            if (!StringUtils.equals(dapDataModel.getModelAdmin(), dapDataModel.getCreateBy())) {
                modelUserAccreditService.saveCreateBy(dapDataModel.getId(), dapDataModel.getCreateBy());
            }
        }
        return aBoolean;
    }

    /**
     * 根据系统id获得集合
     *
     * @param sysId
     * @return
     */
    @Override
    public List<DapDataModel> getModelListById(String sysId) {

        List<DapDataModel> dapDataModels = dapDataModelMapper
                .selectList(new LambdaQueryWrapper<DapDataModel>()
                        .eq(DapDataModel::getSystemId, sysId)
                        .eq(DapDataModel::getBmStatus, BmStatus.ON.getCode())
                        .orderByAsc(DapDataModel::getName));
        return dapDataModels;
    }

    /**
     * 模型详情
     *
     * @param modelId
     * @return
     */
    @Override
    public DapDataModel getModelById(String modelId) {

        DapDataModel dapDataModel = getBaseMapper().selectById(modelId);
        return dapDataModel;
    }

    /**
     * 删除模型
     *
     * @param modelId
     * @return
     */
    @Override
    @Transactional(rollbackFor = {DapException.class, RuntimeException.class})
    @Deprecated
    public ResultBean<Boolean> deleteModel(String modelId) {

        // 获得模型信息
        DapDataModel model = dapDataModelMapper.selectById(modelId);
        if (model.getBmStatus().equals(BmStatus.DELETE)) {
            return ResultBean.error("当前模型已经删除！请刷新页面");
        }
       /* // 如果是公共模型
        if (model.getCommonModel()) {
            dapDataModelMapper.update(null, new LambdaUpdateWrapper<DapDataModel>()
                    .eq(DapDataModel::getBmStatus, BmStatus.ON.getCode())
                    .eq(DapDataModel::getSystemId, model.getSystemId())
                    .set(DapDataModel::getIsUseCommonModel, 0));
            // 删除外键关系
            foreignKeyService.getBaseMapper().delete(new LambdaQueryWrapper<DapDataModelTableForeignKey>()
                    .eq(DapDataModelTableForeignKey::getReferenceModelId, modelId)
                    .or().eq(DapDataModelTableForeignKey::getModelId, modelId));
        }*/
        DapDataModel dapDataModel = new DapDataModel();
        dapDataModel.setId(modelId);
        dapDataModel.setBmStatus(BmStatus.DELETE);
        boolean update = updateById(dapDataModel);
        // 增加清除模型分组
        if (update) {
            groupService.deleteByModelId(false, modelId);
            modelUserAccreditService.deleteByModelId(false, modelId);
        }
        return ResultBean.ok(update);
    }

    /**
     * 修改模型
     *
     * @param vo
     */
    @Override
    @Transactional(rollbackFor = {DapException.class, RuntimeException.class})
    public ResultBean<Boolean> modifyModel(DataModelInsertVo vo) {

        // 更新模型表
        DapDataModel old = this.getModelById(vo.getDapDataModel().getId());
        if (old.getBmStatus().equals(BmStatus.DELETE)) {
            return ResultBean.error("当前模型已删除！请重新刷新页面！");
        }
        // 数据库为mysql且为业务系统类型情况下，查询模型是否有发布记录
        DapSystemDatasource systemDatasource = systemModuleFacade.getDbTypeByDataSourceId(old.getDataSourceId());
        Optional<DapSystemInfo> systemInfo = systemModuleFacade.getSystemInfo(old.getSystemId());
        if (systemDatasource != null && DatabaseType.MYSQL.equals(systemDatasource.getDatabaseType()) &&
                systemInfo.isPresent() && SystemType.BUSINESS_SYS.equals(systemInfo.get().getType())) {
            List<DapDataModelRecord> dapDataModelRecords = dapDataModelRecordService.getBaseMapper().selectList(new LambdaQueryWrapper<DapDataModelRecord>()
                    .eq(DapDataModelRecord::getModelId, vo.getDapDataModel().getId())
                    .eq(DapDataModelRecord::getBmStatus, BmStatus.ON.getCode()));
            if (!dapDataModelRecords.isEmpty() && !old.getIsSyncProd().equals(vo.getDapDataModel().getIsSyncProd())) {
                return ResultBean.error("当前模型已发布，不能更改启用发布同步");
            }
        }
        // 保存模型
        this.updateById(vo.getDapDataModel());
        // 清除之前存在的标签
        dapDataModelSystemTagMiddleService.updateModelTag(vo.getDapDataModel().getId(), 0);
        Boolean aBoolean = true;
        // 判断标签列表是否为空
        if (!vo.getSysTags().isEmpty()) {
            // 插入新增的标签
            aBoolean = dapDataModelSystemTagMiddleService.insertModelTag(vo.getSysTags(), vo.getDapDataModel().getId(), 0);
        }
        // 增加模型管理者的变更操作
        DapDataModel dapDataModel = vo.getDapDataModel();
        if (dapDataModel != null) {
            modelUserAccreditService.changUserAccreditByModelSaveOrUpdate(dapDataModel.getId(), dapDataModel.getModelAdmin());
        }
        return ResultBean.ok(aBoolean);
    }

    /**
     * 根据模型id获取系统下的模型（不排己）
     * 1.找到模型中的系统id
     * 2.验证系统id有效性
     * 3.根据系统id获取模型列表，不排除自己
     *
     * @param modelId 模型id
     * @return 获取模型对应的系统下的公共不排自的模型数据
     */
    @Override
    public List<DapDataModel> findCommonModelsByModelId(String modelId) {
        DapSystemInfo systemInfo = findSystemInfoByModelId(modelId).orElseThrow(() -> new DapException("获取系统信息失败"));
        List<DapDataModel> models = list(bmLambdaQuery().eq(DapDataModel::getSystemId, systemInfo.getId()));
        Set<String> shareModelIds = shareTableService.matchShareModels(models.stream().map(DapDataModel::getId).collect(Collectors.toSet()));
        return models.stream().filter(m -> shareModelIds.contains(m.getId())).collect(Collectors.toList());
    }

    /**
     * 获取模型id对应的系统信息
     *
     * @param modelId 模型id
     * @return 获取模型id对应的系统信息
     */
    @Override
    public Optional<DapSystemInfo> findSystemInfoByModelId(String modelId) {
        DapDataModel model = getByIdAsset(modelId, "获取模型信息失败");
        String systemId = model.getSystemId();
        return systemModuleFacade.getSystemInfo(systemId);
    }

    /**
     * 根据模型id判断是否为数仓系统
     *
     * @param modelId 模型id
     * @return true为是 false否 systemInfo
     */
    @Override
    public Tuple2<Boolean, DapSystemInfo> isDwSystemByModelId(String modelId) {
        DapSystemInfo systemInfo = findSystemInfoByModelId(modelId).orElseThrow(() -> new DapException("模型id获取系统信息失败！！"));
        SystemType sysType = systemInfo.getType();
        // 是否为数仓模型、对应数仓模型的分组和非数仓分组
        boolean isDwSystem = SystemType.WAREHOUSE_SYS.equals(sysType);
        return Tuples.of(isDwSystem, systemInfo);
    }

    /**
     * 根据模型id获取对应的数据库类型
     *
     * @param modelId 模型id
     * @return 数据库类型 当数据源为文件系统时为空
     */
    @Override
    public Tuple2<DapDataModel, Optional<DatabaseType>> findModelAndDbTypeByModelId(String modelId) {
        DapDataModel model = getByIdAsset(modelId, "当前模型不存在！！");
        Optional<DapSystemDatasource> sysDataSourceById = systemModuleFacade.getSysDataSourceById(model.getDataSourceId());
        DapSystemDatasource datasource = sysDataSourceById.orElse(null);
        DatabaseType dbType = null;
        if (datasource != null) {
            dbType = datasource.getDatabaseType();
        }
        return Tuples.of(model, Optional.ofNullable(dbType));
    }

    /**
     * 分别获取模型下的表、字段、索引、分区的列表信息
     *
     * @param modelId 模型id
     * @param newest  是否为最新数据
     * @return 结果集
     */
    @Override
    public Tuple4<List<DapDataModelTable>, List<DapDataModelTableField>, List<DapDataModelSubordinateKey>,
            List<DapDataModelTablePartition>> queryModelInfosByModelAndNewest(String modelId, boolean newest) {
        List<DapDataModelTable> tables = Collections.emptyList();
        List<DapDataModelTableField> fields = Collections.emptyList();
        List<DapDataModelSubordinateKey> indexes = Collections.emptyList();
        List<DapDataModelTablePartition> partitions = Collections.emptyList();
        ExecutorService pool = null;
        try {
            pool = Executors.newFixedThreadPool(4, (r -> new Thread(r, "获取模型信息")));
            Future<List<DapDataModelTable>> tablesF = pool.submit(() -> tableService.findAllByPageByModelIdAndNewest(modelId, newest));
            Future<List<DapDataModelTableField>> fieldsF = pool.submit(() -> fieldService.findAllByPageByModelIdAndNewest(modelId, newest));
            Future<List<DapDataModelSubordinateKey>> indexsF = pool.submit(() -> indexService.getBaseMapper(DapDataModelSubordinateKeyMapper.class).findByModelIdAndNewest(modelId, newest));
            Future<List<DapDataModelTablePartition>> partioF = pool.submit(() -> partitionService.getBaseMapper(DapDataModelTablePartitionMapper.class).findByModelIdAndNewest(modelId, newest));
            tables = tablesF.get();
            fields = fieldsF.get();
            indexes = indexsF.get();
            partitions = partioF.get();
        } catch (InterruptedException | ExecutionException e) {
            log.error("多线程异步查询被中断！！", e);
        } finally {
            if (pool != null) {
                pool.shutdown();
            }
        }
        return Tuples.of(tables, fields, indexes, partitions);
    }

    @Override
    public List<DapDataModel> matchName(String name) {
        if (StringUtils.isBlank(name)) {
            return list(bmLambdaQuery());
        }
        List<DapDataModel> list = list(bmLambdaQuery().like(DapDataModel::getName, name));
        return list;
    }

    @Override
    public IPage<TableScoreDto> getTableScoreList(SearchTableScoreVo vo) {

        Page<DapDataModel> page = new Page<>(vo.getPageNum(), vo.getPageSize());
        IPage<TableScoreDto> tableScoreDtoIPage = dapDataModelMapper.selectTableScoreList(page, vo);
        return tableScoreDtoIPage;
    }
}
